package cn.whuc.spark

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

object Demo_CategoryTop10_4_01 {
  def main(args: Array[String]): Unit = {
    // 1 创建sparkContext
    val sc: SparkContext = new SparkContext(
      new SparkConf()
        .setMaster("local[*]")
        .setAppName(" ")
    )

    // 2 编写代码
    // 2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_37_2019-07-17 00:00:02_手机_-1_-1_null_null_null_null_3
    //2019-07-17_95_26070e87-1ad7-49a3-8fb3-cc741facaddf_48_2019-07-17 00:00:10_null_16_98_null_null_null_null_19

    /*
      思路4 使用累加器实现
      累加器 输入 cid和类型
            输出 Category样例类对象的map集合
     */
    var acc = new HotCategory01Accumulator()

    sc.register(acc)

    val datas: RDD[String] = sc.textFile("input/user_visit_action.txt")

    datas.foreach(line=>{
      val strings: Array[String] = line.split("_")
      if (strings(6)!="-1"){
        // 点击事件
        acc.add(strings(6),"click")
      }else if(strings(8)!="null"){
        val ids = strings(8).split(",")
        ids.map(id=>{
          acc.add(id,"order")
        })
      }else if(strings(10)!="null"){
        val ids = strings(10).split(",")
        ids.map(id=>{
          acc.add(id,"pay")
        })
      }
    })

    val value: mutable.Map[String, Category01] = acc.value

    // 排序
    val result: List[Category01] = value.map(_._2).toList.sortWith((l, r) => {
      if (l.clickcnt > r.clickcnt) {
        true
      } else if (l.clickcnt == r.clickcnt) {
        if (l.ordercnt > r.ordercnt) {
          true
        } else if (l.ordercnt == r.ordercnt) {
          if (l.paycnt > r.paycnt) {
            true
          } else {
            false
          }
        } else {
          false
        }
      } else {
        false
      }
    }).take(10)

    result.foreach(println)

    sc.stop()
  }

}
/*
  样例类：
    类似于JavaBean
    关键字 case 自动带有getter和setter方法，自动实现equals、tostring和hashcode等方法
    其对象为单例对象 singleton
 */
case class Category01(cid:String,var clickcnt:Int,var ordercnt:Int,var paycnt:Int)

/*
  累加器对象
  输入：cid和 id类型
  输出：map[cid,样例类对象]
 */

class HotCategory01Accumulator extends AccumulatorV2[(String,String),mutable.Map[String,Category01]] {
  private val map = mutable.Map[String,Category01]()

  override def isZero: Boolean = map.isEmpty

  override def copy(): AccumulatorV2[(String, String), mutable.Map[String, Category01]] = new HotCategory01Accumulator()

  override def reset(): Unit = map.clear()

  override def add(v: (String, String)): Unit = {
    val cid: String = v._1
    val actionType: String = v._2
    val c: Category01 = map.getOrElse(cid, Category01(cid, 0, 0, 0))
    if(actionType=="click"){
      c.clickcnt+=1
    }else if(actionType=="order"){
      c.ordercnt+=1
    }else{
      c.paycnt+=1
    }
    map.update(cid,c)
  }

  override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, Category01]]): Unit = {
    val map1: mutable.Map[String, Category01] = map
    val map2: mutable.Map[String, Category01] = other.value
    map2.foreach{
      case (cid,c)=>{
        val newc: Category01 = map1.getOrElse(cid, Category01(cid, 0, 0, 0))
        newc.clickcnt+=c.clickcnt
        newc.ordercnt+=c.ordercnt
        newc.paycnt+=c.paycnt
        map1.update(cid,newc)
      }
    }

  }

  override def value: mutable.Map[String, Category01] = map
}

